package io.xxx.omni.oss.listener

import com.baomidou.mybatisplus.extension.kotlin.KtQueryWrapper
import com.google.gson.JsonObject
import io.xxx.omni.oss.common.gson
import io.xxx.omni.oss.domain.Document
import io.xxx.omni.oss.repository.mapper.DocumentMapper
import org.apache.pulsar.client.api.*
import org.springframework.beans.factory.DisposableBean
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
import org.springframework.stereotype.Component
import java.util.concurrent.TimeUnit
import java.util.regex.Pattern

@Component
class PersistenceListener : ApplicationRunner, DisposableBean {

    @Autowired
    private lateinit var pulsarClient: PulsarClient

    @Autowired
    private lateinit var documentMapper: DocumentMapper

    private lateinit var consumer: Consumer<String>

    override fun run(args: ApplicationArguments) {
        consumer = pulsarClient.newConsumer(Schema.STRING)
            .negativeAckRedeliveryDelay(5000, TimeUnit.SECONDS)
            .topicsPattern(Pattern.compile("persistent://public/default/document-.*"))
            .subscriptionName("oss-sub")
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
            .negativeAckRedeliveryDelay(5, TimeUnit.SECONDS)
            .subscriptionType(SubscriptionType.Shared)
            .enableRetry(true)
            .deadLetterPolicy(
                DeadLetterPolicy
                    .builder()
                    .maxRedeliverCount(10)
                    .build()
            )
            .messageListener { consumer, msg ->
                consumer.acknowledge(msg)
                try {
                    val map = gson.fromJson(String(msg.data), JsonObject::class.java)
                    val storeId = map["storeId"].asString
                    val containsData = map["containsData"].asBoolean
                    val document = gson.fromJson(map["document"].asString, Document::class.java)

                    val wrapper = KtQueryWrapper(Document::class.java)
                        .eq(Document::storeId, storeId)
                        .eq(Document::sn, document.sn)
                    if (!containsData)
                        wrapper.select(Document::sn, Document::modified)
                    val oldDocument = documentMapper.selectOne(wrapper)
                    if (oldDocument != null && document > oldDocument) {
                        document.id = oldDocument.id
                        documentMapper.updateById(document)
                    } else {
                        documentMapper.insert(document)
                    }
                } catch (e: Exception) {
                    consumer.negativeAcknowledge(msg)
                }
            }
            .subscribe()
    }

    override fun destroy() {
        consumer.close()
    }
}